package com.yjy.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

@Component
public class MyDelayConsumer implements SmartInitializingSingleton {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void afterSingletonsInstantiated() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay-group");
        consumer.setNamesrvAddr("192.168.145.128:9876");
        try {
            consumer.subscribe("delay-topic", "*");// 可订阅多个tag，但是一个消息只能有一个tag
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    MessageExt messageExt = list.get(0);
                    try {
                        logger.info("接收到消息，模拟处理失败 date:{}, 次数：{}, msgId:{}", new Date(), messageExt.getReconsumeTimes(), messageExt.getMsgId());
                        throw new RuntimeException("测试失败");
                    } catch (Exception e) {
                        setDelayTimeLevel(messageExt);
                        if (messageExt.getReconsumeTimes() > 5) {
                            logger.info("消费重试超过5次，强制消费成功 date:{}, 次数：{}, msgId:{}", new Date(), messageExt.getReconsumeTimes(), messageExt.getMsgId());
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            consumer.start();
            logger.info("consumer启动成功");
        } catch (MQClientException e) {
            logger.error("消费者订阅消息失败，error：" + e);
        }
    }

    /**
     * 主动设置延时级别不生效！原因未知
     * @param messageExt
     */
    // messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    // level=0级表示不延时，level=1表示1级延时，level=2表示2级延时，以此类推。
    // 处理频率为5/5/10/10/10/30/60/60/60/60
    private void setDelayTimeLevel(MessageExt messageExt) {
        switch (messageExt.getReconsumeTimes()) {
            case 0:
                messageExt.setDelayTimeLevel(1);
                break;
            case 1:
                messageExt.setDelayTimeLevel(1);
                break;
            case 2:
                messageExt.setDelayTimeLevel(2);
                break;
            case 3:
                messageExt.setDelayTimeLevel(2);
                break;
            case 4:
                messageExt.setDelayTimeLevel(3);
                break;
            default:
                messageExt.setDelayTimeLevel(4);
                break;
        }
    }
}
